package com.feidee.fd.sml.algorithm.component.preprocess
import com.feidee.fdspark.transformer.{ColSelector, ExceptionalValuesHandler, MetaStorage, ModelType}
import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
import org.apache.spark.sql.DataFrame

import scala.collection.mutable.ArrayBuffer

/**
  * @Author songhaicheng
  * @Date 2019/6/17 15:07
  * @Description 填充字符串列缺失值预处理组件
  * @Reviewer
  */
case class HandleExceptionalValuesParam(
                                     override val input_pt: String,
                                     override val output_pt: String,
                                     override val hive_table: String,
                                     override val flow_time: String,
                                     override val inputCol: String,
                                     override val outputCol: String,
                                     override val modelPath: String,
                                     override val preserveCols: String,
                                     inputColDataType:String,
                                     operator:String,
                                     defineExceptional: String,
                                     solution: String) extends PreprocessorParam {
  def this() = this(null, null, null, null, "input", "output", null, null, null, null,null,null)

  override def verify(): Unit = {
    super.verify()
    require(tool.isNotNull(inputColDataType), "param inputColDataType can't be null")
    require(solution!=null,"param solution can't be null")
  }

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("defineExceptional" -> defineExceptional)
    map += ("solution" -> solution)
    map += ("operator"->operator)
    map += ("inputColDataType"->inputColDataType)
    map
  }
}


class HandleExceptionalValues extends AbstractPreprocessor[HandleExceptionalValuesParam] {

  override def setUp(param: HandleExceptionalValuesParam, data: DataFrame): Array[PipelineStage] = {
    val handler = new ExceptionalValuesHandler()
      .setInputCol(param.inputCol)
      .setOutputCol(param.outputCol)
      .setInputColDataType(param.inputColDataType)
    if(param.solution!=null){
      handler.setSolution(param.solution)
    }
    if(tool.isNotNull(param.operator)){
      handler.setOperator(param.operator)
    }
    if (param.defineExceptional != null){
      handler.setDefineExceptional(param.defineExceptional)
    }
    Array(handler)
  }

}

object HandleExceptionalValues {

  def apply(paramStr: String): Unit = {
    new HandleExceptionalValues()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    HandleExceptionalValues(args(0))
  }
}
